package com.sly.fuxi.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author SLY
 * @date 2025/2/17
 */
@Slf4j
@Component
public class KafkaCousumer {

    /**
     * 接收消息
     *
     * @param record 消息
     * @param ack    ack
     * @author SLY
     * @date 2025/2/17
     */
    @KafkaListener(topics = "sly-001", groupId = "slyGroup", concurrency = "1")
    public void receiveKafka(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            String value = record.value();
            log.info("receiveKafka:{}", value);
            ack.acknowledge();
        } catch (Exception e) {
            ack.acknowledge();
            log.error("receiveKafka:", e);
        }
    }


}
